Happy to see this added. As discussed in #9260, there is a potential to propagate backwards the information contained in PartialSuccess responses from OTLP exports.

I worry about the code complexity introduced to have "success error" responses, meaning error != nil but the interpretation being success. However, this is what it will take to back-propagate partial successes, we want callers to see success with metadata about the number of rejected points if possible. Great to see this, thanks @evan-bradley.

As discussed in open-telemetry/oteps#238, it would be useful for setting the correct otel.outcome label, for callers to have access to the underlying gRPC and/or HTTP error code. Thanks!

This PR was marked stale due to lack of activity. It will be closed in 14 days.

This PR was marked stale due to lack of activity. It will be closed in 14 days.

Codecov Report

Attention: Patch coverage is 95.45455% with 4 lines in your changes missing coverage. Please review.

Project coverage is 92.23%. Comparing base (5726563) to head (55ea049).
Report is 179 commits behind head on main.

@@            Coverage Diff             @@
##             main    #9041      +/-   ##
==========================================
+ Coverage   92.22%   92.23%   +0.01%     
==========================================
  Files         409      411       +2     
  Lines       19134    19222      +88     
==========================================
+ Hits        17646    17730      +84     
- Misses       1127     1131       +4     
  Partials      361      361              

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

Overall feedback for me is that we just want to add more "things". I don't want to block this, but I also feel that we should make sure we add only the things we are 100% sure about it.

Right now we're only adding HTTP/gRPC status code propagation, everything else is marked as unimplemented and still in the design stage. Do you have concerns with HTTP/gRPC status codes, or the way they're communicated?

I really like the MultiErr (current Error) approach and the APIs related to that. It is a nice way to encapsulate fanout errors.

I played with that, but the issue I encountered is that we may want to read these error types at the receiver level, when there may be multiple fanouts in the pipeline. So you could see something like the following:

I don't see how we pull out the joined errors without traversing the whole error tree by repeatedly calling Unwrap. Having a container type to keep track of all the errors just seems simpler.

I see now, I mistook what you said for the Uber multierr package. Glad you like it.

I am not sure if the ErrorData is the right thing though. Is this only for "Transport" errors? I would not return that bool but instead have a enum or a way to say "IsHttp" or "isGrpc" then have "AsHttp" or "AsGrpc" name to be discussed.

This is meant to hold information related to anything the component would like to communicate upstream. Right now we're starting with HTTP and gRPC status codes, but we would like to add retry information, partial success information, etc. in the future. The goal is to start small and iterate as we design the additional options.

This is meant to hold information related to anything the component would like to communicate upstream. Right now we're starting with HTTP and gRPC status codes, but we would like to add retry information, partial success information, etc. in the future. The goal is to start small and iterate as we design the additional options.

Can you please respond to the rest of the comments and see how the design of the "ErrorData" would look like?

Can you please respond to the rest of the comments and see how the design of the "ErrorData" would look like?

Sure thing. The other questions were based on the first question, so I wanted to make sure we were on the same page there if possible.

I am not sure if the ErrorData is the right thing though. Is this only for "Transport" errors? I would not return that bool but instead have a enum or a way to say "IsHttp" or "isGrpc" then have "AsHttp" or "AsGrpc" name to be discussed.

1. Enum: We could do a list of enums, e.g. [IS_RETRYABLE, IS_GRPC], then expose this with a method like Types() []Type. Since an error can contain multiple error types this would have to be a list instead of just a single enum value.
2. IsHTTP + AsHTTP: This essentially conveys the same information as HTTPStatus() (code int, isHTTP bool), so from a functionality standpoint it's not an issue. I would rename IsHTTP to HasHTTP though, since errors can contain multiple types of data. I'm not sure I see the benefit of retrieving this information across two methods, though, what's wrong with the boolean?

Enum: We could do a list of enums, e.g. [IS_RETRYABLE, IS_GRPC], then expose this with a method like Types() []Type. Since an error can contain multiple error types this would have to be a list instead of just a single enum value.
IsHTTP + AsHTTP: This essentially conveys the same information as HTTPStatus() (code int, isHTTP bool), so from a functionality standpoint it's not an issue. I would rename IsHTTP to HasHTTP though, since errors can contain multiple types of data. I'm not sure I see the benefit of retrieving this information across two methods, though, what's wrong with the boolean?

Before answering any of these questions, need to better understand how "correctly" (recommended, or whatever is the right term) chained consumers should error? For me to know the "right" design, I need to understand better the usage. Here are some examples, and we can debate on them what is the correct way (eventually document this). For these examples assume to use logs:

First Example

A -> B (drops logs with invalid trace identifiers) -> C -> D -> E (exports to Loki)

For a request with 10 logs:

• B finds 2 "logs" that are not valid, for example "trace.id" is 34 hex characters.
• E tries to export the logs and 2 out of the 8 remaining (assuming B dropped the 2 invalid logs) are reject with 400 http request.

Can you please tell me exactly which component should propagate/append errors and how?

Second Exmple

                       / -> E - > F
        / ->  C -> D -
A -> B -               \ -> G -> H
        \
         \ -> I -> J -> K

For a request with 10 logs:

• F fails to send 4 logs with 429 http request.
• H fails to send 2 logs with 400 http request.
• D drops 2 logs because of invalid trace identifiers (trace.id too large).
• K fails to send 1 logs with INVALID_ARGUMENT grpc status code.

Can you please tell me exactly which component should propagate/append errors and how?

Thank you for the in-depth examples and illustrations. Without going too deep into the implementation details of how this happens, here's how I see each example playing out. I'm assuming both pipelines are synchronous, as any asynchronous processing will cut error propagation at that point.

First example

A -> B (drops logs with invalid trace identifiers) -> C -> D -> E (exports to Loki)

1. B notes that 2 items failed out of 10 total items.
2. E takes the 400 response with 2/8 records failing and returns an error without any retry information indicating the response code and partial success response.
3. The error is passed upstream and reaches B. Since B also experienced a partial success, it creates a new error object indicating a partial success and adds this to the error it received. We can discuss the exact mechanism for this separately, but for now I'm assuming that we have a single error object that is continually passed up the chain to aggregate individual errors like this.
4. Somewhere between B recording its partial success and A receiving the error, partial success data is aggregated to indicate 4/10 records failed for this pipeline branch, along with any explanation messages from E or B. This is done by a common library shared by components.
5. Assuming A is an OTLP receiver, it ignores the downstream code and returns a 200-equivalent status code with a partial success message indicating 4/10 records failed.

Second example

                       / -> E - > F
        / ->  C -> D -
A -> B -               \ -> G -> H
        \
         \ -> I -> J -> K

1. D drops the two logs and notes this for later when it receives a response from its downstream consumer(s).
2. F returns both partial success information with 4/8 logs failing, and returns that it received a 429 error code. If it considers this retryable, it returns retry information. We assume F is not an OTLP exporter since 429 codes and partial successes aren't compatible in OTLP.
3. H returns an error indicating partial success with 2/8 logs failing, and a 400 error code without any retry information assuming it is using OTLP.
4. D records its partial success information. An internal library could aggregate the top-level count as 6/10 logs failing, choosing the larger number of failed records for the top-level count. All three errors could be captured inside the OTLP partial success payload. All three errors are kept inside the error container in case an upstream receiver uses a protocol that communicates more detailed information about partial successes.
5. K records that 1/10 logs failed to send and returns an error indicating the status code without retry information, assuming it communicates over OTLP.
6. The error is returned to A. If it is configured (likely using a common library) to retry errors within the Collector, it retries the pipeline branch that goes to F since F requested a retry. The branches for H and K will not be retried. If it is configured to return the error to the caller and uses OTLP, it ignores the downstream code and returns a 200-equivalent status code with a partial success message indicating 6/10 failing logs (the highest number) with error information from D, F, H, and K inside the error body.

it retries the pipeline branch that goes to F since F requested a retry. The branches for H and K will not be retried.

How would receiver A do this? It doesn't have the ability when calling ConsumeLogs to specify which path the data should take.

My gut is that the receiver should not retry data if it receives any PartialSuccess or PermanentError data from the error.

How would receiver A do this? It doesn't have the ability when calling ConsumeLogs to specify which path the data should take.

The necessary pipeline topology information would be contained within the error container and would be passed in the context.Context object when the receiver calls the ConsumeLogs function. This would be baked into a shared library so each receiver doesn't have to implement this logic.

My gut is that the receiver should not retry data if it receives any PartialSuccess or PermanentError data from the error.

I think it's ultimately up to the exporter to determine what combinations of options it supports. It's possible that exporter F uses a protocol where the backend supports idempotent submissions and retryable partial successes are possible.

That said, if receiver A is an OTLP receiver and has been configured to pass retry information upstream from the Collector, then it will need to determine whether to retry or return a partial success message since these are mutually exclusive in OTLP. In this case, not retrying seems like the safest course of action, though there could be some nuances here we could discuss in the future.

The necessary pipeline topology information would be contained within the error container and would be passed in the context.Context object when the receiver calls the ConsumeLogs function. This would be baked into a shared library so each receiver doesn't have to implement this logic.

If we want to support retries, let's actually configure the "retries" in the components that support them and potentially add more components that can do retries. If you want crazy ideas, let's push them with the request vs supporting dynamic routing.

If we want to support retries, let's actually configure the "retries" in the components that support them and potentially add more components that can do retries. If you want crazy ideas, let's push them with the request vs supporting dynamic routing.

In general I would agree that retries should be done within the component that needs to do retries, but the goal here is more so to configure scraping receivers that aren't responding to a request. For example, the filelog receiver could throttle reading based on throttling information from downstream endpoints: open-telemetry/opentelemetry-collector-contrib#20864. The reason we need to have routing like this is to make sure we don't replay requests on exporters that have had successful submissions while retrying exporters that want a retry.

cc @dmitryax since you have the most context here.

@bogdandrutu As discussed offline, here are some code snippets showing roughly how the above examples would work using something similar to the proposed API in this PR. I've tried to keep them as short as possible while still being nearly-runnable code. Note that a lot of this still hasn't been implemented within this PR.

Example 1:

// A -> B (drops logs with invalid trace identifiers) -> C -> D -> E (exports to Loki)

// Shared library consumption function akin to consumerretry, fanoutconsumer, etc.
func ConsumeLogs(ctx context.Context, ld plog.Logs) error {
	err := processorB(ctx, ld)

	cerr := consumererror.ErrorContainer{}

	if errors.As(err, &cerr) {
		errs := cerr.Errors()

		partialCount := 0

		for _, e := range errs {
			count, ok := e.Partial()

			if ok {
				partialCount = partialCount + count
			}
		}

		if partialCount > 0 {
			cerr.SetPartialSuccessAggregate(partialCount)
		}
	}
}

func receiverA(ctx context.Context, ld plog.Logs) error {
	err := ConsumeLogs(ctx, ld)

	cerr := consumererror.ErrorContainer{}

	if errors.As(err, &cerr) {
		if cerr.IsPartialSuccess() {
			// Ignore 400 error code because OTLP partial successes use 200 status codes
			return http.Response{code: 200, body: fmt.Sprintf("Partial success: %d out of %d logs failed", cerr.PartialCount(), ld.LogRecordCount())}
		}
	}
}

func processorB(ctx context.Context, ld plog.Logs) error {
	failedItems := 2
	processorErr := consumererror.New(
		errors.New("Logs had trace IDs that failed validation"),
		consumererror.WithPartial(failedItems)
	)

	err := exporterE(ctx, ld)

	if err != nil {
		return consumererror.Combine(err, processorErr)
	}

	return processorErr
}

func exporterE(ctx context.Context, ld plog.Logs) error {
	return consumererror.New(err,
		consumererror.WithPartial(2)
		consumererror.WithHTTPStatus(400),
	  )
}

Example 2:

//                        / -> E - > F
//         / ->  C -> D -
// A -> B -               \ -> G -> H
//         \
//          \ -> I -> J -> K

func receiverA(ctx context.Context, ld plog.Logs) error {
	err := processorB(ctx, ld)

	cerr := consumererror.ErrorContainer{}

	if errors.As(err, &cerr) {
		if cerr.ShouldRetry() {
			ctx := ctx.WithValue("otelcol-retry-info", cerr.Retry())	
			for i := 0; i < maxRetries; i++ {
				err := processorB(ctx, ld)
				if err != nil {
					// check error and continue with retry
				}

				return nil
			}

			return http.Response{code: 400, body: "could not submit data"}
		} else {
			return http.Response{code: cerr.HttpStatusCode(), body: fmt.Sprintf("Partial success: %d out of %d logs failed", cerr.PartialCount(), ld.LogRecordCount())}
		}
	}
}

func fanoutConsumerB(ctx context.Context, ld plog.Logs) error {
	exporters := []consumer.ConsumeLogsFunc{processorD, exporterK}
	var allErrs error
	for _, e := range exporters {
		err := e(ctx, ld)

		if err != nil {
			allErrs := consumererror.Combine(allErrs, err)
		}
	}

	// TODO: Aggregate partial success counts if present

	return allErrs
}

func processorB(ctx context.Context, ld plog.Logs) error {
	return fanoutConsumerB(ctx, ld)
}

func fanoutConsumerD(ctx context.Context, ld plog.Logs) error {
	exporters := []consumer.ConsumeLogsFunc{exporterF, exporterH}
	var allErrs error
	for _, e := range exporters {
		err := e(ctx, ld)

		if err != nil {
			allErrs := consumererror.Combine(allErrs, err)
		}
	}

	// TODO: Aggregate partial success counts if present

	return allErrs
}

func processorD(ctx context.Context, ld plog.Logs) error {
	failedItems := 2
	processorErr := consumererror.New(
		errors.New("Logs had trace IDs that failed validation"),
		consumererror.WithPartial(failedItems)
	)


	err := fanoutConsumerD(ctx, ld)

	if err != nil {
		return consumererror.Combine(err, processorErr)
	}

	return processorErr
}

func exporterF(ctx context.Context, ld plog.Logs) error {
	return consumererror.New(err,
		consumererror.WithPartial(4)
		consumererror.WithHTTPStatus(429),
		consumererror.WithRetry(
			consumerrerror.WithRetryDelay(10 * time.Second),
		),
	  )
}

func exporterH(ctx context.Context, ld plog.Logs) error {
	return consumererror.New(err,
		consumererror.WithPartial(2)
		consumererror.WithHTTPStatus(400),
	  )
}

func exporterK(ctx context.Context, ld plog.Logs) error {
	return consumererror.New(err,
		consumererror.WithPartial(1)
		consumererror.WithGRPCtatus(codes.InvalidArgument),
	  )
}

We had a meeting on 2024-08-27 with the following outcomes:

• I will merge the ErrorContainer and Error types into a single Error type from the user's perspective so both reads and writes occur on the same type.
• Aggregating things like status codes and partial success counts needs to be accounted for early on, since components reading this data only want a single value.
• Retries are a complex issue that we need to carefully consider before moving forward. We will still be leaving them out of this initial version, but the APIs may change if we determine that we don't need some of the sophistication required for functionality like retrying individual pipeline branches.

Discussion on this PR should be moved to #11085

I'm closing this in favor of #11085. Thanks everyone for your thoughts and reviews.